package css.sword.annotation;


import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.util.Map;
import java.util.Properties;

/**
 * 
 *
 * @project 深圳智能政务办公系统
 * @package zhbg.oaxt.common.utils
 * @file OAkafka.java 创建时间:2018年10月24日下午10:41:23
 * @title 标题（要求能简洁地表达出类的功能和职责）
 * @description 描述（简要描述类的职责、实现方式、使用注意事项等）
 * @copyright Copyright (c) 2018 中国软件与技术服务股份有限公司
 * @company 中国软件与技术服务股份有限公司
 * @module 模块: 模块名称
 * @author  曹敏
 * @reviewer 审核人
 * @version 1.0
 * @history 修订历史（历次修订内容、修订人、修订时间等）
 *
 */
public class OAkafka implements Runnable{
    /**参数*/
    private Map<String, Object> map;
    /**参数*/
    private String topics;  
    /**日志*/
     public static final Logger logger = LoggerFactory.getLogger(OAkafka.class);
    /**
     * 
     *@constructor 构造方法
     *@param map 参数
     **@param topics 参数
     */
    public OAkafka(Map<String, Object> map,String topics) {
        this.map =map;
        this.topics=topics;
    }
    
    /**
     * 
     *@constructor 构造方法
     */
    public OAkafka() {
    }

    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年10月25日上午9:38:54
     *@return 返回值
     *@author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public Map<String, Object> getMap() {
        return map;
    }
   
    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年10月31日下午7:53:15
     *@return  返回值
     *@author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public String getTopics() {
        return topics;
    }
    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年10月31日下午7:53:20
     *@param topics  返回值
     *@author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public void setTopics(String topics) {
        this.topics = topics;
    }

    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年10月25日上午9:39:37
     *@param  map  参数
     *@author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public void setMap(Map<String, Object> map) {
        this.map = map;
    }
    
    
    
    /**
     *  
     *@name    读取配置文件
     *@description 相关说明
     *@time    创建时间:2018年10月31日下午6:00:50
     *@return  返回值
     *@author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public static Properties getprops(){
        final Properties props=new Properties() ;
        try{
            final String config="producer.properties";
            final InputStream in= OAkafka.class.getClassLoader().getResourceAsStream(config);
            props.load(in);
            in.close();
        }catch(Exception e){
            e.printStackTrace();
        }
        return props;
    }     
    /**
     * 
     *@name    异步发送消息
     *@description 相关说明
     *@time    创建时间:2018年11月3日下午12:50:13
     *@param topics 参数
     *@param map    参数
     *@author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public static void sendkafka(Map<String,Object> map,String topics){
        try {
            if(null==OAKafkaManger.producer) {
                final KafkaProducer<String, String> producer = new KafkaProducer<String, String>(getprops());
                OAKafkaManger.producer = producer;
            }
            OAKafkaManger.producer.send(new ProducerRecord<String, String>(topics, JSONObject.toJSONString(map)));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }    
    }
    
    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年11月3日下午1:22:30
     * @author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    @Override
    public void run() {
        sendkafka(map,topics);
    }    
}
